/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package cassandrawindow;
//import com.vcc.hector.config.cosistenlevel.VccSimpleConsistentLevel;
import java.io.IOException;
import java.lang.Long;
import java.math.BigInteger;
import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;
import me.prettyprint.cassandra.model.ConfigurableConsistencyLevel;
import me.prettyprint.cassandra.model.HColumnImpl;
import me.prettyprint.cassandra.serializers.LongSerializer;
import me.prettyprint.cassandra.serializers.StringSerializer;
import me.prettyprint.cassandra.service.CassandraHostConfigurator;
import me.prettyprint.cassandra.service.ColumnSliceIterator;
import me.prettyprint.cassandra.service.FailoverPolicy;
import me.prettyprint.cassandra.service.OperationType;
import me.prettyprint.cassandra.service.ThriftCluster;
import me.prettyprint.cassandra.service.ThriftKsDef;
import me.prettyprint.hector.api.Cluster;
import me.prettyprint.hector.api.ConsistencyLevelPolicy;
import me.prettyprint.hector.api.HConsistencyLevel;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.beans.ColumnSlice;
import me.prettyprint.hector.api.beans.CounterSlice;
import me.prettyprint.hector.api.beans.CounterSuperRow;
import me.prettyprint.hector.api.beans.CounterSuperRows;
import me.prettyprint.hector.api.beans.CounterSuperSlice;
import me.prettyprint.hector.api.beans.HColumn;
import me.prettyprint.hector.api.beans.HCounterColumn;
import me.prettyprint.hector.api.beans.HCounterSuperColumn;
import me.prettyprint.hector.api.beans.HSuperColumn;
import me.prettyprint.hector.api.beans.OrderedRows;
import me.prettyprint.hector.api.beans.Row;
import me.prettyprint.hector.api.beans.Rows;
import me.prettyprint.hector.api.beans.SuperRow;
import me.prettyprint.hector.api.beans.SuperSlice;
import me.prettyprint.hector.api.ddl.ColumnFamilyDefinition;
import me.prettyprint.hector.api.ddl.ComparatorType;
import me.prettyprint.hector.api.ddl.KeyspaceDefinition;
import me.prettyprint.hector.api.exceptions.HectorException;
import me.prettyprint.hector.api.factory.HFactory;
import me.prettyprint.hector.api.mutation.Mutator;
import me.prettyprint.hector.api.query.ColumnQuery;
import me.prettyprint.hector.api.query.CounterQuery;
import me.prettyprint.hector.api.query.MultigetSliceQuery;
import me.prettyprint.hector.api.query.MultigetSuperSliceCounterQuery;
import me.prettyprint.hector.api.query.MultigetSuperSliceQuery;
import me.prettyprint.hector.api.query.QueryResult;
import me.prettyprint.hector.api.query.RangeSlicesQuery;
import me.prettyprint.hector.api.query.SliceCounterQuery;
import me.prettyprint.hector.api.query.SliceQuery;
import org.apache.cassandra.thrift.TokenRange;
import org.apache.cassandra.thrift.Column;
import vcc.util.adsllog.StringSplitLogUtil;

public class CassandraAction {
    private static String host_cs       = "";
    private static String port_cs       = "";
    private static String clusterName   = "";
    private ThriftCluster cassandraCluster;
    private CassandraHostConfigurator cassandraHostConfigurator;
    private static StringSerializer stringSerializer = StringSerializer.get();
    private static LongSerializer longSerializer = LongSerializer.get();
    private ConsistencyLevelPolicy consistenlevelpolicy ;
    private Mutator<String> mutator;
    private static Keyspace keyspaceOperator;
    
    private static String V3_BAN_DATE          = "v3_ban_date_CS";
    private static String V3_CAM_DATE          = "v3_cam_date_CS";
    private static String V3_CAM_HOURLY_CPC    = "v3_cam_hourly_CS";
    private static String   V3_DOMAIN_DATE     = "v3_domain_date_cs";
    private static String V3_ZONE_DATE         = "v3_zone_date_CS";
    private static String V3_BAN_HOURLY_CPC    = "v3_ban_hourly_CS";
    

    
    CassandraAction(String nameCluster ,String host,String port) {
        //setupCase();
        host_cs = host;
        port_cs = port;
        clusterName = nameCluster;        
    }

    public void setupCase(){
        cassandraHostConfigurator = new CassandraHostConfigurator(host_cs + ":" + port_cs);
        cassandraCluster = new ThriftCluster(clusterName, cassandraHostConfigurator);        
    }
    
    public void initReadWirte(String keyspace){
        /*
         * chi dung cong 9161 va consistentLevel cho mobile 
        */        
        CassandraHostConfigurator cassandraHostConfigurator = new CassandraHostConfigurator(
                    "192.168.23.115:9161");
        cassandraHostConfigurator.setMaxActive(50);
        Cluster cluster = HFactory.getOrCreateCluster(clusterName, cassandraHostConfigurator);        
        //keyspaceOperator = HFactory.createKeyspace(keyspace, cluster);      
                                
        VccSimpleConsistentLevel consistentLevel=new VccSimpleConsistentLevel();  
        consistentLevel.get(OperationType.READ);
        //keyspaceOperator.setConsistencyLevelPolicy(consistentLevel);
        
        keyspaceOperator = HFactory.createKeyspace(keyspace, cluster,consistentLevel);                
        mutator = HFactory.createMutator(keyspaceOperator, StringSerializer.get());  
          
        
    }

    public void createColumnFamily(String keyspace ,String columnFamily)
    {
        ColumnFamilyDefinition cfDef = HFactory.createColumnFamilyDefinition(keyspace, columnFamily);
        cassandraCluster.addColumnFamily(cfDef);
    }
    public void truncateColumnFamily(String keyspace ,String columnFamily)
    {
        cassandraCluster.truncate(keyspace, columnFamily);
        
    }
    public void dropColumnFamily(String keyspace ,String columnFamily){
        cassandraCluster.dropColumnFamily(keyspace,columnFamily);
    }

    public void dropKeyspaces (String keyspace)
    {
        cassandraCluster.dropKeyspace(keyspace);
    }

    public void describeKeyspaces(String keyspace) throws Exception {
        List<TokenRange> ring = cassandraCluster.describeRing(keyspace);
        Debug.println(ring.toString());
    }

    public KeyspaceDefinition checkExistKeyspace(String keyspace){        
        Cluster cluster = HFactory.getOrCreateCluster(clusterName, host_cs + ":" + port_cs);
        KeyspaceDefinition keyspaceDef = cluster.describeKeyspace(keyspace);
        return keyspaceDef;
    }
        
    public void setKeyspace (String keyspace,String ColumnFamilyName)
    {
        /*
            CassandraHostConfigurator cassandraHostConfigurator = new CassandraHostConfigurator(
                    "192.168.23.84:9160,192.168.23.85:9160,192.168.23.86:9160");
            cassandraHostConfigurator.setMaxActive(200);
            //so thread in pool ( trong file config)            
            Cluster cluster = HFactory.getOrCreateCluster("Adlog cluster", cassandraHostConfigurator);
        */            
            Cluster cluster = HFactory.getOrCreateCluster(clusterName, host_cs + ":" + port_cs);
            ColumnFamilyDefinition cfDef = HFactory.createColumnFamilyDefinition(keyspace, ColumnFamilyName, ComparatorType.BYTESTYPE);
            int replicationFactor = 0;                      
            KeyspaceDefinition newKeyspace = HFactory.createKeyspaceDefinition(keyspace, ThriftKsDef.DEF_STRATEGY_CLASS,  replicationFactor, Arrays.asList(cfDef));
            // Add the schema to the cluster.           
            cluster.addKeyspace(newKeyspace, true);
    }

    public static void insertColumFamily(String keyspace ,String nameColum)
    {
        Cluster cluster = HFactory.getOrCreateCluster(clusterName, host_cs + ":" + port_cs);
        ColumnFamilyDefinition cfDef = HFactory.createColumnFamilyDefinition(keyspace,nameColum,ComparatorType.BYTESTYPE);        
        
    }

    public void insertValueCassandra(String nameColum,String key,String name,String value)
    {
      try {
            mutator.insert(key, nameColum, HFactory.createStringColumn(name, value));            
            //mutator.insert("quyen", nameColum, HFactory.createStringColumn("adress", "HANOI"));                        
        } catch (HectorException e) {
            e.printStackTrace();
            Debug.println("co loi trong qua trinh insert du lieu");
        }
    }

    public void insertValueCassandraV2(String nameColum,String key,String name,String value)
    {            
        try {                                    
            mutator.addInsertion(key, nameColum, HFactory.createStringColumn(name, value));

        } catch (HectorException e) {
            e.printStackTrace();
            Debug.println("co loi trong qua trinh insert du lieu");
        }
    }

    public String getValueCassandara(String keyspace,String nameColum, String key,String name)
    {
        String reVal ="";
        //get value
        //Cluster cluster = HFactory.getOrCreateCluster(clusterName, host_cs + ":" + port_cs);
        //Keyspace keyspaceOperator = HFactory.createKeyspace(keyspace, cluster);
        ColumnQuery<String, String, String> columnQuery = HFactory.createStringColumnQuery(keyspaceOperator);
        columnQuery.setColumnFamily(nameColum).setKey(key).setName(name);
        QueryResult<HColumn<String, String>> result = columnQuery.execute();
        
        System.out.println("Read HColumn from cassandra: " + result.get());
        System.out.println("Verify on CLI with:  get Keyspace1.Users['jsmith'] ");
        
        return reVal;

    }
    public static void deleteData(String keyspace ,String nameColum, String key) {
            Cluster cluster = HFactory.getOrCreateCluster(clusterName, host_cs + ":" + port_cs);
            Keyspace keyspaceOperator = HFactory.createKeyspace(keyspace, cluster);
            try {
              //Create a mutator object for this keyspace using utf-8 encoding
                 Mutator<String> mutator = HFactory.createMutator(keyspaceOperator, StringSerializer.get());
                //Use the mutator object to delete row
                mutator.delete(key, nameColum, null, stringSerializer);
                //close connection
                cluster.getConnectionManager().shutdown();

            } catch (Exception ex) {
                System.out.println("Error encountered while deleting data!!");
                ex.printStackTrace() ;
            }
        }

    public void getSliceQueryFromCounterFamily(String key,String columnFamily){                
        long start  = 1340944200000L;
        long finish = 1340949600000L;

        SliceCounterQuery<String, Long> query  = HFactory.createCounterSliceQuery(keyspaceOperator, stringSerializer,LongSerializer.get());                

        query.setKey(key);
        query.setColumnFamily(columnFamily);
        query.setRange(start, finish, false, 10);
        
        List<HCounterColumn<Long>> columnList = query.execute().get().getColumns();
        
        for(HCounterColumn colum : columnList){
            System.out.println("Count:" + colum.getValue());
        }        
        //return results;
    }

    public void getOneValueInColumnFamily(String key,String columnFamily,String nameColumn)
    {
        CounterQuery<String, String> query = HFactory.createCounterColumnQuery(keyspaceOperator, stringSerializer,stringSerializer);
        query.setColumnFamily(columnFamily).setKey(key).setName(nameColumn);
        HCounterColumn<String> counter = query.execute().get();
        System.out.println("Count:" + counter.getValue());
    }

    public long getCountClickByIntervalTime(String key, String columnFamily,long startTime,long finisTime)
    {
        long retValue = 0;
        //int count = (int) ((finisTime - startTime) / (15 * 60 * 1000)) + 1;
        int count = 4;//moi 15 phut
        SliceCounterQuery<String, Long> query  = HFactory.createCounterSliceQuery(keyspaceOperator, stringSerializer,LongSerializer.get());
        
        query.setKey(key);
        query.setColumnFamily(columnFamily);
        query.setRange(startTime, finisTime, false, count);        

        List<HCounterColumn<Long>> columnList = query.execute().get().getColumns();

        for(HCounterColumn colum : columnList){
            retValue = retValue + colum.getValue();
        }

        return retValue;
    }

    public String getIpByCampaign(String key, String columnFamily)
    {                        
            StringBuilder sb = new StringBuilder();                                    
            long start = System.currentTimeMillis();
            
            ColumnSliceIterator<String, String, String> it = new
                ColumnSliceIterator<String, String, String>(HFactory.createSliceQuery(keyspaceOperator, StringSerializer.get(), StringSerializer.get(), StringSerializer.get())
                .setColumnFamily(columnFamily)
                .setKey(key), null, (String)null, false);
            
            //long counter     = 0;                        
            String[] arr = new String[3];
            
            while (it.hasNext()) {                
                HColumn<String, String> next = it.next();                                                
                arr = next.getName().split("_");
                
                // tranh truong hop rong va -1
                if((arr[0].length()>1)&&(!arr[0].equals("-1")))
                {                    
                    sb.append("(").append("'").append(arr[0]).append("'").append(",").append("'").append(arr[1]).append("'").append(",").append("'").append(key).append("'").append(")").append(",");
                    //counter++;
                }
                // insert DB 100 lan 
                /*if(((counter % 100) == 0) && (counter != 0)){
                    String relStr = sb.toString();
                    String valInsert = relStr.substring(0, (relStr.length()-1));
                    mysqldatamanager.insertDataCampaignIp(valInsert);
                    sb.setLength(0);
                }*/
            }
            
            //xu li cho du lieu cuoi
            String relStr = sb.toString();
            String valInsert = relStr.substring(0, (relStr.length()-1));
            return valInsert;
            //mysqldatamanager.insertDataCampaignIp(valInsert);
    }

    public void getDomaindateMobile(String key, String columnFamily, MysqlDataManager mysql,MysqlDataManagerMain mysqlMain) throws InterruptedException, ParseException{               
        MysqlDataManager67 mysql67 = new MysqlDataManager67();
        StringBuilder sb = new StringBuilder();        
        Map<String, Integer> priceByDay = new HashMap<>();
        List validBanner = new ArrayList();        
        //gia doi voi moi domain
        priceByDay = mysqlMain.getPriceByDomain();
        //lay nhung domain hop le        
        validBanner = mysqlMain.getValidDomain();           
                     
        MultigetSuperSliceCounterQuery<String, String, String> multigetSuperSliceCounterQuery = HFactory.createMultigetSuperSliceCounterQuery(keyspaceOperator, stringSerializer, stringSerializer, stringSerializer);        
        multigetSuperSliceCounterQuery.setColumnFamily(columnFamily)
                //.setRange("", "", false, 5000)                
                .setKeys(key);
        String stkey = null;        
        boolean checkBreak  = true;
        Iterator<CounterSuperRow<String, String, String>> iterators;
        
        do {
            multigetSuperSliceCounterQuery.setRange(stkey, null, false, 1000);
            iterators = multigetSuperSliceCounterQuery.execute().get().iterator();
            
            while (iterators.hasNext()) {
                try {
                    CounterSuperRow<String, String, String> nextSuperRow = iterators.next();
                    
                    CounterSuperSlice<String, String> superSlice = nextSuperRow.getSuperSlice();
                    List<HCounterSuperColumn<String, String>> superColumns = superSlice.getSuperColumns();
                    //superColumns.size();
                    if (superColumns.isEmpty()) {
                        checkBreak  =   false;
                    }
                    if (superColumns.size() == 1)
                    {
                         checkBreak  =   false;
                    }

                    for (HCounterSuperColumn<String, String> counterSuperColumn : superColumns) {
                        //BannerZoneCvInfo bannerZoneCvInfo = new BannerZoneCvInfo();
                        
                        if (stkey != null && stkey.equals(counterSuperColumn.getName()))
                        {
                            continue;
                        } else{
                            //System.out.println(counterSuperColumn.getName());
                        }
                        stkey = counterSuperColumn.getName();
                        
                        //bannerZoneCvInfo.setBannerId(Long.valueOf(nextSuperRow.getKey()));
                        //bannerZoneCvInfo.setZoneId(Long.valueOf(counterSuperColumn.getName()));
                        List<HCounterColumn<String>> columns = counterSuperColumn.getColumns();
                        long clickCount = 0;
                        
                        long viewCount = 0;
                        
                        for (HCounterColumn<String> hCounterColumn : columns) {
                            if (hCounterColumn.getName().equals("click")) {
                                clickCount = hCounterColumn.getValue();
                                //System.out.println(hCounterColumn.getValue());
                                //bannerZoneCvInfo.setClick(hCounterColumn.getValue());
                                continue;
                            }
                            if (hCounterColumn.getName().equals("view")) {
                                viewCount = hCounterColumn.getValue();
                                //System.out.println(hCounterColumn.getValue());
                                //bannerZoneCvInfo.setView(hCounterColumn.getValue());
                                continue;
                            }
                        }
                        //zoneCvInfoList.add(bannerZoneCvInfo);
                        String[] arr = new String[5];
                        arr = counterSuperColumn.getName().split("_");
                        //System.out.println(arr.length);
                        /*
                         * co nhung domain = default va campaignid =-1 phai loai di
                         * arr : campaignid_bannerid_domain
                         */
                        //Date currentDay = java.sql.Date.valueOf(key);
                        if(arr.length == 3 && (arr[0] == null ? "-1" != null : !arr[0].equals("-1")) && (arr[2] == null ? "defaultdomain" != null : !arr[2].equals("defaultdomain"))
                            &&Integer.valueOf(arr[0]) < 9999999 && Integer.valueOf(arr[1]) < 999999999 && Integer.valueOf(arr[0]) > 0 && Integer.valueOf(arr[1]) > 0&&arr[2].length() < 30)
                        {                           
                            if(arr[0].trim().length() > 0 && arr[1].trim().length() >0&& arr[2].trim().length() > 0)
                            {

                                    if(priceByDay.containsKey(arr[2]))
                                    {                                       
                                        int price = priceByDay.get(arr[2]);
                                        String strResult = "('" + arr[0] + "'," + arr[1] + ",'" + arr[2] + "'," +clickCount +"," + clickCount + "," + viewCount +"," + viewCount +",'" + arr[2] +"',0,1,'"+ arr[2] + "','" + arr[2] +"','" + key + "',0," + price*(clickCount)+ ")";
                                        sb.append(strResult).append(",");
                                        
                                    } else{
                                        //System.out.println(clickCount);
                                        //String strResult = "('" + arr[0] + "'," + arr[1] + ",'" + arr[2] + "'," +clickCount +"," + clickCount + "," + viewCount +"," + viewCount +",CRC32('" + arr[2] +"'),0,0,'"+ arr[2] + "',CRC32('" + arr[2] +"')"+ ",DATE('" + key + "'),0,0)";
                                        String strResult = "('" + arr[0] + "'," + arr[1] + ",'" + arr[2] + "'," +clickCount +"," + clickCount + "," + viewCount +"," + viewCount +",'" + arr[2] +"',0,1,'"+ arr[2] + "','" + arr[2] +"','" + key + "',0,0)";
                                        sb.append(strResult).append(",");
                                    }

                                }                                   
                        }else{
                            continue;
                        }
                        
                        if(sb.toString().length() > 100)
                        {                             
                            //xu li cho du lieu cuoi
                            //MysqlDataManager mysqlTemp = new MysqlDataManager();
                            //connect to DB 6.7
                            //MysqlDataManager67 mysql67 = new MysqlDataManager67();
                            String relStr = sb.toString();
                            String valInsert = relStr.substring(0, (relStr.length()-1));
                            mysql67.insertDataDomainData(valInsert);
                            sb.setLength(0);
                        }
                        
                    }                    
                } catch (Exception exp) {
                    System.out.println("error:"+exp.getMessage());
                    continue;
                }
            }
            //} while (columns.size() > 1 && counter < 800);
        } while (checkBreak);
        
        /*
        if(sb.toString().length() > 1)
        {
            //xu li cho du lieu cuoi
            //connect to DB 6.7
            //MysqlDataManager67 mysql67 = new MysqlDataManager67();
            String relStr = sb.toString();
            String valInsert = relStr.substring(0, (relStr.length()-1));           
            HandleImportData hid = new HandleImportData();

            String pathFile="";
            try {
                
                pathFile = hid.outToCsvDomain(valInsert);
                Debug.println("pathFile:"+pathFile);
            } catch (IOException ex) {
                ex.printStackTrace();
            }                
            try {
                 //get sys_date_cpc
                String strDate =  mysql67.getSysDateCPC();
                
                //drop partition
                //mysql67.dropPartition(CONSTANT.V3_DOMAIN_DATE_CS, strDate);
                MysqlDataManager67 mysql672 = new MysqlDataManager67(false);                
                //insert data
                mysql672.beforeLoadData();
                mysql672.dropPartition(CONSTANT.V3_DOMAIN_DATE_CS, strDate);
                mysql672.loadToV3DomainDateFromFile(pathFile);
                //close connetion
                mysql672.shutdownConnection();
                mysql67.shutdownConnection();
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        }*/
        
    }

    public void getDomaindateCPM(String key, String columnFamily, MysqlDataManager mysql,MysqlDataManagerMain mysqlMain) throws InterruptedException{

        MysqlDataManagerMainCPM mysqlcpm = new MysqlDataManagerMainCPM();

        StringBuilder sb = new StringBuilder();
        Map<String, Integer> fraudclicks = new HashMap<>();
        Map<String, Integer> priceByDay = new HashMap<>();
        List validBanner = new ArrayList();
        fraudclicks = mysql.getAllFraudClickByDay(key);
        
        priceByDay = mysqlMain.getPriceByDomainCPC();
        //lay nhung domain hop le
        validBanner = mysqlMain.getValidDomain();
        //lay nhung campaign thuoc cpc plus
        List CampaignCPM = new ArrayList();
        CampaignCPM = mysqlMain.getCampaignCPM();
        //lay gia cpcplus doi voi moi domain
        //Map<String, Integer> PriceByDomainCPCPlus = new HashMap<>();
        //PriceByDomainCPCPlus = mysqlMain.getPriceByDomainCPCPlus();

        MultigetSuperSliceCounterQuery<String, String, String> multigetSuperSliceCounterQuery = HFactory.createMultigetSuperSliceCounterQuery(keyspaceOperator, stringSerializer, stringSerializer, stringSerializer);

        multigetSuperSliceCounterQuery.setColumnFamily(columnFamily)
                //.setRange("", "", false, 5000)
                .setKeys(key);
        String stkey = null;
        boolean checkBreak  = true;
        Iterator<CounterSuperRow<String, String, String>> iterators;

        do {
            multigetSuperSliceCounterQuery.setRange(stkey, null, false, 1000);
            iterators = multigetSuperSliceCounterQuery.execute().get().iterator();

            while (iterators.hasNext()) {
                try {
                    CounterSuperRow<String, String, String> nextSuperRow = iterators.next();

                    CounterSuperSlice<String, String> superSlice = nextSuperRow.getSuperSlice();
                    List<HCounterSuperColumn<String, String>> superColumns = superSlice.getSuperColumns();
                    //superColumns.size();
                    if (superColumns.isEmpty()) {
                        checkBreak  =   false;
                    }
                    if (superColumns.size() == 1)
                    {
                         checkBreak  =   false;
                    }

                    for (HCounterSuperColumn<String, String> counterSuperColumn : superColumns) {
                        //BannerZoneCvInfo bannerZoneCvInfo = new BannerZoneCvInfo();

                        if (stkey != null && stkey.equals(counterSuperColumn.getName()))
                        {
                            continue;
                        } else{
                            //System.out.println(counterSuperColumn.getName());
                        }
                        stkey = counterSuperColumn.getName();

                        //bannerZoneCvInfo.setBannerId(Long.valueOf(nextSuperRow.getKey()));
                        //bannerZoneCvInfo.setZoneId(Long.valueOf(counterSuperColumn.getName()));
                        List<HCounterColumn<String>> columns = counterSuperColumn.getColumns();
                        long clickCount = 0;

                        long viewCount = 0;

                        for (HCounterColumn<String> hCounterColumn : columns) {
                            if (hCounterColumn.getName().equals("click")) {
                                clickCount = hCounterColumn.getValue();
                                //System.out.println(hCounterColumn.getValue());
                                //bannerZoneCvInfo.setClick(hCounterColumn.getValue());
                                continue;
                            }
                            if (hCounterColumn.getName().equals("view")) {
                                viewCount = hCounterColumn.getValue();
                                //System.out.println(hCounterColumn.getValue());
                                //bannerZoneCvInfo.setView(hCounterColumn.getValue());
                                continue;
                            }
                        }
                        //zoneCvInfoList.add(bannerZoneCvInfo);
                        String[] arr = new String[5];
                        arr = counterSuperColumn.getName().split("_");
                        //System.out.println(arr.length);
                        /*
                         * co nhung domain = default va campaignid =-1 phai loai di
                         * arr : campaignid_bannerid_domain
                         */
                        if(arr.length == 3 && (arr[0] == null ? "-1" != null : !arr[0].equals("-1")) && (arr[2] == null ? "defaultdomain" != null : !arr[2].equals("defaultdomain"))&&!arr[0].equals(""))
                        {
                            //neu la campaignCPM
                                if(CampaignCPM.contains(Long.valueOf(arr[0])))
                                {
                                    // neu banner hop le
                                    if(validBanner.contains(arr[2]))
                                    {
                                            if(arr[0].trim().length() > 0 && arr[1].trim().length() >0&& arr[2].trim().length() > 0)
                                            {
                                                //int fraudClick = 0;
                                                if(fraudclicks.containsKey(arr[1]+"_"+arr[2]+"_"+key))
                                                {
                                                    if(priceByDay.containsKey(arr[2]))
                                                    {
                                                        int fraudClick = fraudclicks.get(arr[1]+"_"+arr[2]+"_"+key);

                                                        if((clickCount - fraudClick) >= 0)
                                                        {
                                                            int price = priceByDay.get(arr[2]);
                                                            int cpmRound = Math.round(viewCount/1000);
                                                            String domainCPM = arr[2];                                                            
                                                            String strResult = "('" + arr[0] + "'," + arr[1] + ",'" + arr[2] + "'," +clickCount +"," + viewCount + "," +(clickCount - fraudClick) +"," + viewCount+ "," +(clickCount - fraudClick)+"," + viewCount +",CRC32('" + domainCPM +"'),0,0,'"+ domainCPM + "',CRC32('" + domainCPM +"'),'"+ key + "',0," + price*cpmRound+ ")";
                                                            sb.append(strResult).append(",");
                                                        }
                                                    } else {
                                                            int price = priceByDay.get(arr[2]);
                                                            int cpmRound = Math.round(viewCount/1000);
                                                             String domainCPM = arr[2];                                                            
                                                            String strResult = "('" + arr[0] + "'," + arr[1] + ",'" + arr[2] + "'," +clickCount +"," + viewCount +"," + clickCount+"," + viewCount +"," + clickCount+"," + viewCount+",CRC32('" + domainCPM +"'),0,0,'"+ domainCPM + "',CRC32('" + domainCPM +"'),'"+key + "',0," + price*cpmRound+ ")";
                                                            sb.append(strResult).append(",");
                                                    }
                                                } else{
                                                    if(priceByDay.containsKey(arr[2]))
                                                    {
                                                            int price = priceByDay.get(arr[2]);
                                                            int cpmRound = Math.round(viewCount/1000);
                                                            String domainCPM = arr[2];
                                                            String strResult = "('" + arr[0] + "'," + arr[1] + ",'" + arr[2] + "'," +clickCount +"," + viewCount +"," + clickCount+"," + viewCount +"," + clickCount+"," + viewCount+",CRC32('" + domainCPM +"'),0,0,'"+ domainCPM + "',CRC32('" + domainCPM +"'),'"+key + "',0," + price*cpmRound+ ")";
                                                            sb.append(strResult).append(",");
                                                    } else{
                                                            //System.out.println(clickCount);
                                                            int price = priceByDay.get(arr[2]);
                                                            int cpmRound = Math.round(viewCount/1000);
                                                            String domainCPM = arr[2];
                                                            String strResult = "('" + arr[0] + "'," + arr[1] + ",'" + arr[2] + "'," +clickCount +"," + viewCount + "," + clickCount+"," + viewCount+ "," + clickCount+"," + viewCount +",CRC32('" + domainCPM +"'),0,0,'"+ domainCPM + "',CRC32('" + domainCPM +"'),'"+ key + "',0," + price*cpmRound+ ")";
                                                            sb.append(strResult).append(",");
                                                    }
                                                }
                                                }
                                            }
                                }
                            }else{
                                continue;
                            }

                        if(sb.toString().length() > 20000)
                        {
                            //xu li cho du lieu cuoi
                            String relStr = sb.toString();
                            String valInsert = relStr.substring(0, (relStr.length()-1));
                            mysqlcpm.insertDataDomainData(valInsert);
                            sb.setLength(0);
                        }

                    }
                } catch (Exception exp) {
                    System.out.println("error:"+exp.getMessage());
                    continue;
                }
            }
            //} while (columns.size() > 1 && counter < 800);
        } while (checkBreak);
        if(sb.toString().length() > 1)
        {
            //xu li cho du lieu cuoi
            String relStr = sb.toString();
            String valInsert = relStr.substring(0, (relStr.length()-1));
            mysqlcpm.insertDataDomainData(valInsert);
            sb.setLength(0);
        }
    }


    public void getZoneDate(String key, String columnFamily, MysqlDataManager mysql,MysqlDataManagerMain mysqlMain) throws InterruptedException
    {
        StringBuilder sb = new StringBuilder();
        Map<String, Integer> fraudclicks = new HashMap<>();
        Map<String, Integer> priceByDay = new HashMap<>();
        List validBanner = new ArrayList();
        fraudclicks = mysql.getAllFraudClickByZone(key);
        priceByDay = mysqlMain.getPriceByDomain();
        //lay nhung domain hop le        
        validBanner = mysqlMain.getValidDomain();
        //lay nhung campaign thuoc cpc plus        
        List CampaignCPCPlus = new ArrayList();
        CampaignCPCPlus = mysqlMain.getCampaignCPCPlus();
        
        //lay gia cpcplus doi voi moi domain
        Map<String, Integer> PriceByDomainCPCPlus = new HashMap<>();
        PriceByDomainCPCPlus = mysqlMain.getPriceByDomainCPCPlus();

        MultigetSuperSliceCounterQuery<String, String, String> multigetSuperSliceCounterQuery = HFactory.createMultigetSuperSliceCounterQuery(keyspaceOperator, stringSerializer, stringSerializer, stringSerializer);
        multigetSuperSliceCounterQuery.setColumnFamily(columnFamily)
                //.setRange("", "", false, 5000)
                .setKeys(key);
        String stkey = null;
        boolean checkBreak  = true;
        Iterator<CounterSuperRow<String, String, String>> iterators;

        do {
            multigetSuperSliceCounterQuery.setRange(stkey, null, false, 1000);
            iterators = multigetSuperSliceCounterQuery.execute().get().iterator();

            while (iterators.hasNext()) {
                try {
                    CounterSuperRow<String, String, String> nextSuperRow = iterators.next();
                    CounterSuperSlice<String, String> superSlice = nextSuperRow.getSuperSlice();
                    List<HCounterSuperColumn<String, String>> superColumns = superSlice.getSuperColumns();
                    //superColumns.size();
                    if (superColumns.isEmpty()) {
                        checkBreak  =   false;
                    }
                    if (superColumns.size() == 1)
                    {
                         checkBreak  =   false;
                    }
                    
                    for (HCounterSuperColumn<String, String> counterSuperColumn : superColumns) {
                        //BannerZoneCvInfo bannerZoneCvInfo = new BannerZoneCvInfo();

                        if (stkey != null && stkey.equals(counterSuperColumn.getName()))
                        {
                            continue;
                        } else{
                            //System.out.println(counterSuperColumn.getName());
                        }
                        stkey = counterSuperColumn.getName();

                        //bannerZoneCvInfo.setBannerId(Long.valueOf(nextSuperRow.getKey()));
                        //bannerZoneCvInfo.setZoneId(Long.valueOf(counterSuperColumn.getName()));
                        List<HCounterColumn<String>> columns = counterSuperColumn.getColumns();
                        long clickCount = 0;

                        long viewCount = 0;

                        for (HCounterColumn<String> hCounterColumn : columns) {
                            if (hCounterColumn.getName().equals("click")) {
                                clickCount = hCounterColumn.getValue();
                                //System.out.println(hCounterColumn.getValue());
                                //bannerZoneCvInfo.setClick(hCounterColumn.getValue());
                                continue;
                            }
                            if (hCounterColumn.getName().equals("view")) {
                                viewCount = hCounterColumn.getValue();
                                //System.out.println(hCounterColumn.getValue());
                                //bannerZoneCvInfo.setView(hCounterColumn.getValue());
                                continue;
                            }
                        }                        
                        String[] arr = new String[5];
                        //bannerid_zone_campaignid
                        arr = counterSuperColumn.getName().split("_");
                        
                        if(arr.length == 3)
                        {
                            if(arr[0].trim().length() > 0 && arr[1].trim().length() >0&& arr[2].trim().length() > 0)
                            {
                                if(fraudclicks.containsKey(arr[0]+"_"+arr[2]+"_"+arr[1]+"_"+key))//bannerid_campaignid_zone
                                {
                                        if(priceByDay.containsKey(arr[2]))
                                        {
                                            int fraudClick = fraudclicks.get(arr[0]+"_"+arr[2]+"_"+arr[1]+"_"+key);
                                            if((clickCount - fraudClick) >= 0)
                                            {
                                                String strResult = "('" + arr[2] + "','" + arr[1] + "','" + arr[0] + "'," +clickCount +"," + "0" + "," + "0" + "," + viewCount +",0,'" + key + "',0)";
                                                sb.append(strResult).append(",");
                                            }
                                        }
                                }
                            }
                        }else{
                            continue;
                        }                      

                        if(sb.toString().length() > 20000)
                        {
                            //MysqlDataManager mysql = new MysqlDataManager();
                            //xu li cho du lieu cuoi
                            String relStr = sb.toString();
                            String valInsert = relStr.substring(0, (relStr.length()-1));
                            mysql.insertDataZone(valInsert);
                            sb.setLength(0);
                        }

                    }
                } catch (Exception exp) {
                    System.out.println("error:"+exp.getMessage());
                    continue;
                }
            }
            //} while (columns.size() > 1 && counter < 800);
        } while (checkBreak);
        
        if(sb.toString().length() > 1)
        {
            //MysqlDataManager mysql = new MysqlDataManager();
            //xu li cho du lieu cuoi
            String relStr = sb.toString();
            String valInsert = relStr.substring(0, (relStr.length()-1));
            mysql.insertDataZone(valInsert);
            sb.setLength(0);
        }
    }

    public void getZoneDateCPC(String key, String columnFamily, MysqlDataManager mysql,MysqlDataManagerMain mysqlMain) throws InterruptedException, ClassNotFoundException, InstantiationException, IllegalAccessException
    {
        MysqlDataManager67 mysql67 = new MysqlDataManager67();
        MysqlDataManager mysqlCPC  = new MysqlDataManager();
        
        StringBuilder sb = new StringBuilder();
        Map<String, Integer> fraudclicks = new HashMap<>();                
        fraudclicks = mysql.getAllFraudClickByZone(key);               

        MultigetSuperSliceCounterQuery<String, String, String> multigetSuperSliceCounterQuery = HFactory.createMultigetSuperSliceCounterQuery(keyspaceOperator, stringSerializer, stringSerializer, stringSerializer);
        multigetSuperSliceCounterQuery.setColumnFamily(columnFamily)
                //.setRange("", "", false, 5000)
                .setKeys(key);
        String stkey = null;
        boolean checkBreak  = true;
        Iterator<CounterSuperRow<String, String, String>> iterators;

        do {
            multigetSuperSliceCounterQuery.setRange(stkey, null, false, 1000);
            iterators = multigetSuperSliceCounterQuery.execute().get().iterator();

            while (iterators.hasNext()) {
                try {
                    CounterSuperRow<String, String, String> nextSuperRow = iterators.next();
                    CounterSuperSlice<String, String> superSlice = nextSuperRow.getSuperSlice();
                    List<HCounterSuperColumn<String, String>> superColumns = superSlice.getSuperColumns();
                    //superColumns.size();
                    if (superColumns.isEmpty()) {
                        checkBreak  =   false;
                    }
                    if (superColumns.size() == 1)
                    {
                         checkBreak  =   false;
                    }

                    for (HCounterSuperColumn<String, String> counterSuperColumn : superColumns) {
                        //BannerZoneCvInfo bannerZoneCvInfo = new BannerZoneCvInfo();

                        if (stkey != null && stkey.equals(counterSuperColumn.getName()))
                        {
                            continue;
                        } else{
                            //System.out.println(counterSuperColumn.getName());
                        }
                        stkey = counterSuperColumn.getName();

                        //bannerZoneCvInfo.setBannerId(Long.valueOf(nextSuperRow.getKey()));
                        //bannerZoneCvInfo.setZoneId(Long.valueOf(counterSuperColumn.getName()));
                        List<HCounterColumn<String>> columns = counterSuperColumn.getColumns();
                        long clickCount = 0;

                        long viewCount = 0;

                        for (HCounterColumn<String> hCounterColumn : columns) {
                            if (hCounterColumn.getName().equals("click")) {
                                clickCount = hCounterColumn.getValue();
                                //System.out.println(hCounterColumn.getValue());
                                //bannerZoneCvInfo.setClick(hCounterColumn.getValue());
                                continue;
                            }
                            if (hCounterColumn.getName().equals("view")) {
                                viewCount = hCounterColumn.getValue();
                                //System.out.println(hCounterColumn.getValue());
                                //bannerZoneCvInfo.setView(hCounterColumn.getValue());
                                continue;
                            }
                        }
                        String[] arr = new String[5];
                        //bannerid_zone_campaignid
                        arr = counterSuperColumn.getName().split("_");

                        if(arr.length == 3 && Long.valueOf(arr[0]) < 9999999 && Long.valueOf(arr[1]) < 100000 && Long.valueOf(arr[2]) < 99999999 && Long.valueOf(arr[2]) > 0 && Long.valueOf(arr[1]) >0 && Long.valueOf(arr[0]) > 0)
                        {
                            if(arr[0].trim().length() > 0 && arr[1].trim().length() >0&& arr[2].trim().length() > 0&& 0 < Long.valueOf(arr[0])&& 0 < Long.valueOf(arr[1])&& 0 < Long.valueOf(arr[2]))
                            {
                                if(fraudclicks.containsKey(arr[0]+"_"+arr[2]+"_"+arr[1]+"_"+key))//bannerid_campaignid_zone
                                {                                       
                                    int fraudClick = fraudclicks.get(arr[0]+"_"+arr[2]+"_"+arr[1]+"_"+key);
                                    if((clickCount - fraudClick) >= 0)
                                    {
                                        String strResult = "(" + Integer.valueOf(arr[2]) + "," + Integer.valueOf(arr[1]) + "," + Integer.valueOf(arr[0]) + "," +clickCount +"," + (clickCount - fraudClick)+ "," + (clickCount - fraudClick) + "," + viewCount + ","  +viewCount+ ",'" + key + "',0)";
                                        sb.append(strResult).append(",");
                                    }
                                }else{
                                        String strResult = "(" + Integer.valueOf(arr[2]) + "," + Integer.valueOf(arr[1]) + "," + Integer.valueOf(arr[0]) + "," +clickCount +"," + clickCount + "," + clickCount + "," + viewCount +"," +viewCount+ ",'" + key + "',0)";
                                        sb.append(strResult).append(",");
                                }
                            }
                        }else{
                            continue;
                        }

                        /*if(sb.toString().length() > 2000)
                        {
                            //MysqlDataManager mysql = new MysqlDataManager();
                            //xu li cho du lieu cuoi
                            String relStr = sb.toString();
                            String valInsert = relStr.substring(0, (relStr.length()-1));
                            mysql67.insertDataZoneCPC(valInsert);
                            sb.setLength(0);
                        }*/
                    }
                } catch (Exception exp) {
                    System.out.println("error:"+exp.getMessage());
                    continue;
                }
            }
            //} while (columns.size() > 1 && counter < 800);
        } while (checkBreak);

       //xu li cho du lieu cuoi
        if(sb.toString().length() > 1)
        {
            //MysqlDataManager mysql = new MysqlDataManager();
            //xu li cho du lieu cuoi
            //connect to DB 6.7
            //MysqlDataManager67 mysql67 = new MysqlDataManager67();
            String relStr = sb.toString();
            String valInsert = relStr.substring(0, (relStr.length()-1));
            /*mysql67.insertDataDomainData(valInsert);
            sb.setLength(0);*/
            HandleImportData hid = new HandleImportData();
            String pathFile="";
            try {
                pathFile = hid.outToCsvZone(valInsert);
                Debug.println("pathFile:"+pathFile);
            } catch (IOException ex) {
                ex.printStackTrace();
            }

            try {
                //get sys_date_cpc
                String strDate =  mysql67.getSysDateCPC();
                
                //mysql67.dropPartition(CONSTANT.V3_ZONE_DATE_CS, strDate);
                MysqlDataManager67 mysql672 = new MysqlDataManager67(false);
                mysql672.beforeLoadData();
                mysql672.dropPartition(CONSTANT.V3_ZONE_DATE_CS, strDate);
                mysql672.loadToV3ZoneFromFile(pathFile);
                mysql672.shutdownConnection();
                mysql67.shutdownConnection();
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        }
    }

    public String getFraudClick(String key, long time ,String columnFamily ,String insertDate)
    {       
        StringBuilder sb = new StringBuilder();

        MultigetSuperSliceQuery<String, Long, String ,String> multigetSuperSliceQuery = HFactory.createMultigetSuperSliceQuery(keyspaceOperator, stringSerializer, LongSerializer.get(), stringSerializer, stringSerializer);

        multigetSuperSliceQuery.setColumnFamily(columnFamily)
                //.setRange("", "", false, 5000)
                .setKeys(key);
        //long stkey = 0L;
        //boolean checkBreak  = true;
        long endTime = time;
        multigetSuperSliceQuery.setRange(time, endTime, false, 1000);
        
        Iterator<SuperRow<String, Long, String ,String>> iterators;
        iterators = multigetSuperSliceQuery.execute().get().iterator();
            
            while (iterators.hasNext()) {
                
                try {
                    SuperRow<String, Long, String ,String> nextSuperRow = iterators.next();
                    //System.out.println(nextSuperRow.getKey());
                    SuperSlice<Long, String,String> superSlice = nextSuperRow.getSuperSlice();
                    
                    List<HSuperColumn<Long, String,String>> superColumns = superSlice.getSuperColumns();
                                        
                    for (HSuperColumn<Long, String,String> SuperColumn : superColumns) {
                        
                        //System.out.println(SuperColumn.getName());
                        List<HColumn<String, String>> columns= SuperColumn.getColumns();
                        for (HColumn<String, String> column : columns) {
                            //check null
                            if(column.getName().length() > 1)
                            {
                                String[] arr = new String[100];
                                arr = column.getName().split("\t");
                                //check enough data
                                if((arr[7].length() > 1)&&(arr[23].length() > 1))
                                {
                                    
                                    String insertData = "('" + column.getName() + "', " + key + "," + Integer.valueOf(StringSplitLogUtil.getZoneIdFromLog(arr))+ "," +Integer.valueOf(StringSplitLogUtil.getCampaignIdFromLog(arr))+ "," + Integer.valueOf(StringSplitLogUtil.getBannerIdFromLog(arr))+ ",'" + StringSplitLogUtil.getDomain(arr) +"',"+ 0 +",'" + insertDate +"')";
                                    sb.append(insertData).append(",");
                                }
                                
                            }
                        }
                        
                     }
                } catch (Exception exp) {
                    System.out.println("error:"+exp.getMessage());
                    continue;
                }
            }
        return sb.toString();
    }

    public String getFraudClickDay(String key,String columnFamily ,String insertDate)
    {
        StringBuilder sb = new StringBuilder();

        MultigetSliceQuery<String, String ,String> multigetSliceQuery = HFactory.createMultigetSliceQuery(keyspaceOperator, stringSerializer,stringSerializer, stringSerializer);
        multigetSliceQuery.setColumnFamily(columnFamily);
        multigetSliceQuery.setKeys(key);
        multigetSliceQuery.setRange(null, null, false, 10000);
        QueryResult<Rows<String, String, String>> result = multigetSliceQuery.execute();
        Rows<String, String, String> orderedRows = result.get();
      
        //System.out.println("Contents of rows: \n");
        for (Row<String, String, String> r : orderedRows) {
            List<HColumn<String,String>> listColumns = r.getColumnSlice().getColumns();
            Iterator it = listColumns.iterator();
            try {
                while(it.hasNext())
                {
                    HColumn<String,String> columns = (HColumn<String, String>) it.next();
                    //System.out.println(columns.getName());
                    //System.out.println(columns.getValue());
                    if(columns.getName().length() > 1)
                    {
                        String[] arr = new String[100];
                        arr = columns.getName().split("\t");
                        //check enough data
                        String insertData = "('" + columns.getName() + "', '" + StringSplitLogUtil.getGuidId(arr) + "'," + Integer.valueOf(StringSplitLogUtil.getZoneIdFromLog(arr))+ "," +Integer.valueOf(StringSplitLogUtil.getCampaignIdFromLog(arr))+ "," + Integer.valueOf(StringSplitLogUtil.getBannerIdFromLog(arr))+ ",'" + StringSplitLogUtil.getDomain(arr) +"',"+ 0 +",'" + insertDate +"')";
                        sb.append(insertData).append(",");
                    }
                }
            }catch (Exception exp) {
                Debug.println("error:"+exp.getMessage());
                continue;
            }
        }
        return sb.toString();
    }

    public String getFraudClickDomain(String key,String columnFamily ,String insertDate)
    {
        StringBuilder sb = new StringBuilder();

        MultigetSliceQuery<String, String ,String> multigetSliceQuery = HFactory.createMultigetSliceQuery(keyspaceOperator, stringSerializer,stringSerializer, stringSerializer);
        multigetSliceQuery.setColumnFamily(columnFamily);
        multigetSliceQuery.setKeys(key);
        multigetSliceQuery.setRange(null, null, false, 10000);
        QueryResult<Rows<String, String, String>> result = multigetSliceQuery.execute();
        Rows<String, String, String> orderedRows = result.get();            
            for (Row<String, String, String> r : orderedRows) {
                List<HColumn<String,String>> listColumns = r.getColumnSlice().getColumns();
                Iterator it = listColumns.iterator();
                try {
                    while(it.hasNext())
                    {
                           HColumn<String,String> columns = (HColumn<String, String>) it.next();
                           //System.out.println(columns.getName());
                           //System.out.println(columns.getValue());

                           if(columns.getName().length() > 1)
                            {
                                String[] arr = new String[100];
                                arr = columns.getName().split("\t");
                                //check enough data
                                String insertData = "('" + columns.getName() + "', '" + StringSplitLogUtil.getGuidId(arr) + "'," + Integer.valueOf(StringSplitLogUtil.getZoneIdFromLog(arr))+ "," + Integer.valueOf(StringSplitLogUtil.getCampaignIdFromLog(arr))+ "," + Integer.valueOf(StringSplitLogUtil.getBannerIdFromLog(arr))+ ",'" + StringSplitLogUtil.getDomain(arr) +"',"+ 0 +",'" + insertDate +"')";
                                sb.append(insertData).append(",");
                            }
                    }
                }catch (Exception exp) {
                    System.out.println("error:"+exp.getMessage());
                    continue;
                }
            }
        return sb.toString();
    }

    public String getMN_MB_MT(String date,String bannerid ,String columnFamily)
    {
        String value = "";
        //String key = date+"_"+bannerid;
        String key = "2012-08-10_150912";
        SliceCounterQuery<String, Long> query  = HFactory.createCounterSliceQuery(keyspaceOperator, stringSerializer,LongSerializer.get());
        query.setKey(key);
        query.setColumnFamily(columnFamily);
        query.setRange(null, null, false, 1000);
        List<HCounterColumn<Long>> columnList = query.execute().get().getColumns();

        for(HCounterColumn colum : columnList){
            //retValue = retValue + colum.getValue();
            Debug.println(String.valueOf(colum.getValue()));
            value = value + String.valueOf(colum.getValue())+"_";
        }
        return value;
        
    }

    public String getGeo(String key, String columnFamily,long startTime,long finishTime) throws InterruptedException{
        
        MultigetSuperSliceCounterQuery<String, Long, String> multigetSuperSliceCounterQuery = HFactory.createMultigetSuperSliceCounterQuery(keyspaceOperator, stringSerializer, longSerializer, stringSerializer);
        StringBuilder sb = new StringBuilder();
        
        long MB = 0;
        long MT = 0;
        long MN = 0;
        long OTHER = 0;
        
        multigetSuperSliceCounterQuery.setColumnFamily(columnFamily)
                //.setRange("", "", false, 5000)
                .setKeys(key);
        //String stkey = null;
        //long stkey = 1354700700000;
        long stkey = startTime;        
        boolean checkBreak  = true;
        //Iterator<CounterSuperRow<String, String, String>> iterators;
        Iterator<CounterSuperRow<String, Long, String>> iterators;
        do {
            //MultigetSuperSliceCounterQuery<String, Long, String> multigetSuperSliceCounterQuery = new MultigetSuperSliceCounterQuery<String, Long, String>();
            multigetSuperSliceCounterQuery.setRange(stkey, finishTime, false, 1000);
            
            //multigetSuperSliceCounterQuery.setRange(key, key, checkBreak, count)
            iterators = multigetSuperSliceCounterQuery.execute().get().iterator();

            while (iterators.hasNext()) {
                try {
                    CounterSuperRow<String, Long, String> nextSuperRow = iterators.next();

                    CounterSuperSlice<Long, String> superSlice = nextSuperRow.getSuperSlice();
                    
                    List<HCounterSuperColumn<Long, String>> superColumns = superSlice.getSuperColumns();
                    //superColumns.size();
                    if (superColumns.isEmpty()) {
                        checkBreak  =   false;
                    }
                    if (superColumns.size() == 1)
                    {
                         checkBreak  =   false;
                    }

                    for (HCounterSuperColumn<Long, String> counterSuperColumn : superColumns) {
                        //BannerZoneCvInfo bannerZoneCvInfo = new BannerZoneCvInfo();

                        if (stkey != 0 && stkey == counterSuperColumn.getName())
                        {
                            continue;
                        } else{
                            //System.out.println(counterSuperColumn.getName());
                        }
                        stkey = counterSuperColumn.getName();
                        //System.out.println(stkey);
                        //bannerZoneCvInfo.setBannerId(Long.valueOf(nextSuperRow.getKey()));
                        //bannerZoneCvInfo.setZoneId(Long.valueOf(counterSuperColumn.getName()));
                        List<HCounterColumn<String>> columns = counterSuperColumn.getColumns();                       

                        for (HCounterColumn<String> hCounterColumn : columns) {
                            //System.out.println(hCounterColumn.getName());
                            //System.out.println(hCounterColumn.getValue());
                            
                            if (hCounterColumn.getName() == null ? "0" == null : hCounterColumn.getName().equals("0")) {
                                OTHER = OTHER + hCounterColumn.getValue();
                                //System.out.println(hCounterColumn.getValue());
                                //bannerZoneCvInfo.setClick(hCounterColumn.getValue());
                                continue;
                            }
                            else if (hCounterColumn.getName() == null ? "1" == null : hCounterColumn.getName().equals("1")) {
                                MB = MB + hCounterColumn.getValue();
                                //System.out.println(hCounterColumn.getValue());
                                //bannerZoneCvInfo.setView(hCounterColumn.getValue());
                                continue;
                            }else if (hCounterColumn.getName() == null ? "2" == null : hCounterColumn.getName().equals("2")) {
                                MT = MT + hCounterColumn.getValue();
                                //System.out.println(hCounterColumn.getValue());
                                //bannerZoneCvInfo.setView(hCounterColumn.getValue());
                                continue;
                            }else if (hCounterColumn.getName() == null ? "3" == null : hCounterColumn.getName().equals("3")) {
                                MN = MN + hCounterColumn.getValue();
                                //System.out.println(hCounterColumn.getValue());
                                //bannerZoneCvInfo.setView(hCounterColumn.getValue());
                                continue;
                            }
                        }                                                                     
                    }
                } catch (Exception exp) {
                    System.out.println("error:"+exp.getMessage());
                    continue;
                }
            }
            //} while (columns.size() > 1 && counter < 800);
        } while (checkBreak);
       /* if((MB == 0.00)&(MT == 0.00)&(MN == 0.00)&(OTHER == 0.00))
        {
            Debug.println(key);
        }*/
        sb.append(MB).append("_").append(MT).append("_").append(MN).append("_").append(OTHER);        
        return sb.toString();
    }
    
    
    public void getDataMobile(String key, String columnFamily,String columnFamilyClick,String columnFamilyView,long startTime,long finisTime,String currentDate ) throws InterruptedException, ClassNotFoundException, InstantiationException, IllegalAccessException
    {                             
            int  countInvalidDomain = 0; 
            MysqlDataManager67 mysqlMain = new MysqlDataManager67();
            Map<String,String> dataMobile=new HashMap<>();
            dataMobile = mysqlMain.getDataMobile();
            
            Map<String,String> banneridByDomain= new HashMap<String,String>();
            //get all domain by hanging for each user        
            banneridByDomain = mysqlMain.getAllDomainByBanneridMobile();
            
            //get banner cpc va cpc plus            
            CheckPauseBannerAndCampaign cbc = new CheckPauseBannerAndCampaign();                         
            //get data pause
            HashSet dataPause = new HashSet();
            dataPause = cbc.getDataStatusPause(mysqlMain, key);            
            //get campaign out of date cpc
            List CampaignCPCOutOfBudget = new ArrayList();
            CampaignCPCOutOfBudget = mysqlMain.getCampaignOutBudgetMobile();
            //get campaignid  complete
            List CampaignComplete = new ArrayList();
            CampaignComplete = mysqlMain.getCampaignComplete();
            
            StringBuilder sb = new StringBuilder();                                                            
            ColumnSliceIterator<String, String, String> it = new
                ColumnSliceIterator<String, String, String>(HFactory.createSliceQuery(keyspaceOperator, StringSerializer.get(), StringSerializer.get(), StringSerializer.get())
                .setColumnFamily(columnFamily)
                .setKey(key), null, (String)null, false);
            
            while (it.hasNext()) {                
                HColumn<String, String> next = it.next();                                                
                String keyCombine = next.getName();
                
                try {
                    String[] arr = new String[3];
                    arr = keyCombine.split("_");
                    
                    long countClick = getDataMobileByIntervalTime(keyCombine, columnFamilyClick, startTime, finisTime);                
                    long countView = getDataMobileByIntervalTime(keyCombine, columnFamilyView, startTime, finisTime);
                    //get banner from key                
                    String bannerid     = arr[1];
                    String domain       = "";
                    if(arr.length == 2)
                    {                    
                        domain = "other";                    
                    } else if(arr.length == 3) {                    
                        domain  = arr[2];
                    } 
                    
                    String campaignid   = arr[0];
                                        
                    //check domain valid
                    if(banneridByDomain.containsKey(arr[1])){                              
                        String domainBan = banneridByDomain.get(arr[1]);                                    
                        if(!domainBan.contains(domain)){
                            countInvalidDomain ++;                                    
                            continue;
                        }                                   
                    }
                                                            
                    if(dataMobile.containsKey(bannerid)){
                        String valueMobile  = dataMobile.get(bannerid);
                        String price    = valueMobile.split("_")[0];
                        String typepay  = valueMobile.split("_")[1];
                        //String typead   = valueMobile.split("_")[2];
                        long money = 0;
                        
                        //neu campaign thuoc user cpc da het ngan sach thi ko tinh tien chi thong ke click
                        //hoac neu chua campaign da dung tu hom truoc
                        if(CampaignCPCOutOfBudget.contains(arr[0]) || dataPause.contains(arr[0]) || CampaignComplete.contains(arr[0])){
                           String strResult = "(" + Integer.valueOf(campaignid) + "," + Integer.valueOf(bannerid) + ",'" + domain + "'," +countClick +"," + countClick + "," + countView +"," + countView + ",'" + domain +"','"+ domain + "','" + domain +"','"+ currentDate + "',1,0" +")";
                            sb.append(strResult).append(",");     
                        } else {
                            //neu CPM
                            if(typepay.equals("2")){                           
                                money = (int)((Integer.valueOf(price)*countView)/1000);                                                                        
                            //neu CPC    
                            } else if (typepay.equals("1")) {                         
                                money = (int)((Integer.valueOf(price)*countClick));                                                                            
                            }

                            String strResult = "(" + Integer.valueOf(campaignid) + "," + Integer.valueOf(bannerid) + ",'" + domain + "'," +countClick +"," + countClick + "," + countView +"," + countView + ",'" + domain +"','"+ domain + "','" + domain +"','"+ currentDate + "',1," + money + ")";
                            sb.append(strResult).append(",");                            
                        }                                                                                
                    } 
                    
                } catch (Exception ex ){
                    System.out.println(ex.getMessage());
                }
            }
            
        System.out.println("countInvalidDomain:"+countInvalidDomain);    
        //xu li cho du lieu cuoi        
        if(sb.toString().length() > 1)
        {
            //xu li cho du lieu cuoi
            //connect to DB 6.7
            //MysqlDataManager67 mysql67 = new MysqlDataManager67();
            String relStr = sb.toString();
            String valInsert = relStr.substring(0, (relStr.length()-1));           
            HandleImportData hid = new HandleImportData();

            String pathFile="";
            try {
                
                pathFile = hid.outToCsvDomain(valInsert);
                Debug.println("pathFile:"+pathFile);
            } catch (IOException ex) {
                ex.printStackTrace();
            }                
            try {
                 //get sys_date_cpc
                //String strDate =  mysql67.getSysDateCPC();
                
                //drop partition
                //mysql67.dropPartition(CONSTANT.V3_DOMAIN_DATE_CS, strDate);
                MysqlDataManager mysql86 = new MysqlDataManager();                
                //insert data
                //mysql672.beforeLoadData();
                mysql86.removeDataDomainByDate(currentDate);
                mysql86.loadToV3DomainDateFromFile(pathFile);
                //close connetion
                mysql86.shutdownConnection();
                //mysql67.shutdownConnection();
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        }
        mysqlMain.shutdownConnection();
                        
    }
   
    public long getDataMobileByIntervalTime(String key, String columnFamily,long startTime,long finisTime)
    {
        long retValue = 0;
        //int count = (int) ((finisTime - startTime) / (15 * 60 * 1000)) + 1;
        int count = 96;//moi 15 phut
        SliceCounterQuery<String, Long> query  = HFactory.createCounterSliceQuery(keyspaceOperator, stringSerializer,LongSerializer.get());
        
        query.setKey(key);
        query.setColumnFamily(columnFamily);
        query.setRange(startTime, finisTime, false, count);        
        
        List<HCounterColumn<Long>> columnList = query.execute().get().getColumns();

        for(HCounterColumn colum : columnList){
            retValue = retValue + colum.getValue();
        }

        return retValue;
    }
}
   
   
